package com.ftx.connector.source;

import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.errors.ConnectException;
import org.apache.kafka.connect.source.SourceRecord;
import org.apache.kafka.connect.source.SourceTask;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;

public class ExampleSourceTask extends SourceTask {

    // 声明文件名变量
    private String filename;
    // 声明主题变量
    private String topic;


    @Override
    public String version() {
        return new ExampleSourceConnector().version();
    }
    //任务启动
    @Override
    public void start(Map<String, String> props) {
        // 取配置
        filename = props.get(ExampleSourceConnector.FILE_CONFIG);
        topic = props.get(ExampleSourceConnector.TOPIC_CONFIG);
        if (topic == null) {
            throw new ConnectException("FileStreamSourceTask config missing topic setting");
        }
    }
    //需要发送到kafka的数据。
    @Override
    public List<SourceRecord> poll() throws InterruptedException {
        Thread.sleep(1000 * 10);
        String value = "文件名：" + filename + " 主题变量：" + topic + "当前时间戳：" + System.currentTimeMillis();
        SourceRecord sourceRecord = new SourceRecord(
                null,
                null,
                "AI-Proxy-Request-Topic33",
                null,
                Schema.STRING_SCHEMA,
                "KEY",
                Schema.STRING_SCHEMA,
                value);
        List<SourceRecord> list = new ArrayList<>();
        list.add(sourceRecord);
        return list;
    }



    //任务停止
    @Override
    public void stop() {

    }
}
